package org.xiaoyu.rpc.client;

import cn.hutool.core.collection.CollectionUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.xiaoyu.rpc.Loadbalancer;
import org.xiaoyu.rpc.RpcContext;
import org.xiaoyu.rpc.common.exception.RpcException;
import org.xiaoyu.rpc.remoting.transport.cllient.future.HashedWheelTimer;
import org.xiaoyu.rpc.remoting.transport.cllient.future.NamedThreadFactory;
import org.xiaoyu.rpc.remoting.transport.cllient.future.RpcFuture;
import org.xiaoyu.rpc.common.constant.RpcError;
import org.xiaoyu.rpc.common.extension.ExtensionLoader;
import org.xiaoyu.rpc.common.url.URL;
import org.xiaoyu.rpc.registry.Discover;
import org.xiaoyu.rpc.registry.DiscoverFactory;
import org.xiaoyu.rpc.remoting.packet.AppResponse;
import org.xiaoyu.rpc.remoting.packet.Invocation;
import org.xiaoyu.rpc.remoting.transport.cllient.ChannelProvider;
import org.xiaoyu.rpc.remoting.transport.cllient.RpcMessageChecker;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;

import static org.xiaoyu.rpc.config.PropertiesConfig.properties;

/**
 * Created By Have
 * 2021/7/14 15:09
 */
@Slf4j
public class ClientHandler {
    private long RPC_TIMEOUT = 3000;
    private static final DiscoverFactory discoverFactory = ExtensionLoader.getExtensionLoader(DiscoverFactory.class).getExtension("nacos");
    private static final Loadbalancer loadBalancer = ExtensionLoader.getExtensionLoader(Loadbalancer.class).getExtension("weight");
    private static final HashedWheelTimer timer = new HashedWheelTimer(new NamedThreadFactory("future", true));
    private static final ThreadPoolExecutor executor = new ThreadPoolExecutor(16, 100, 300, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), new NamedThreadFactory("Sender", false));
    static {
        timer.start();
    }

    public Object sendRequest(Invocation invocation) {
        try {
            // 需要发现服务
            Discover discover = discoverFactory.createDiscover(properties);
            List<URL> urls = discover.lookupService(invocation.getUrl());
            if (CollectionUtil.isEmpty(urls)) {
                throw new RpcException("No Remoting Service is available");
            }
            URL url = loadBalancer.loadBalance(urls, invocation);
            InetSocketAddress inetSocketAddress = new InetSocketAddress(url.getHost(), url.getPort());
            Channel channel = ChannelProvider.get(inetSocketAddress);
            RpcFuture rpcFuture = RpcFuture.newFuture(channel, invocation, 300);
            executor.submit(new Runnable() {
                @Override
                public void run() {
                    log.info("服务[{}]所在地址[{}]:[{}]", invocation.getServiceName(), inetSocketAddress.getAddress(), inetSocketAddress.getPort());
                    channel.writeAndFlush(invocation)
                            .addListener(new ChannelFutureListener() {
                                @Override
                                public void operationComplete(ChannelFuture future) throws Exception {
                                    if(future.isSuccess()) {
                                        log.info("客户端发送消息:[{}]", invocation);
                                        RpcFuture.sent(channel, invocation);
                                    } else {
                                        log.error("客户端发送消息:[{}]失败, 原因:[{}]", invocation, future.cause());
                                    }
                                }
                            });
                }
            });

            while (!rpcFuture.isDone()) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException ignored) {
                }
            }
            AppResponse response = (AppResponse) rpcFuture.get();
            RpcMessageChecker.check(invocation, response);
            return response.getData();
        } catch (Exception e) {
            log.error("Client Request Error:", e.getMessage(), e);
            throw new RuntimeException("Client Error");
        }
    }

    public RpcFuture sendRequestAsync(Invocation invocation) {
        try {
            // 需要发现服务
            Discover discover = discoverFactory.createDiscover(properties);
            List<URL> urls = discover.lookupService(invocation.getUrl());
            if (CollectionUtil.isEmpty(urls)) {
                log.error("Service is not available!");
                return null;
            }
            URL url = loadBalancer.loadBalance(urls, invocation);
            if (url == null) {
                throw new RpcException("No Remoting Service is available");
            }
            InetSocketAddress inetSocketAddress = new InetSocketAddress(url.getHost(), url.getPort());
            Channel channel = ChannelProvider.get(inetSocketAddress);
            RpcFuture rpcFuture = RpcFuture.newFuture(channel, invocation, 300);
            executor.submit(new Runnable() {
                @Override
                public void run() {
                    log.info("服务[{}]所在地址[{}]:[{}]", invocation.getServiceName(), inetSocketAddress.getAddress(), inetSocketAddress.getPort());
                    channel.writeAndFlush(invocation)
                            .addListener(new ChannelFutureListener() {
                                @Override
                                public void operationComplete(ChannelFuture future) throws Exception {
                                    if(future.isSuccess()) {
                                        log.info("客户端发送消息:[{}]", invocation);
                                        RpcFuture.sent(channel, invocation);
                                    } else {
                                        log.error("客户端发送消息:[{}]失败, 原因:[{}]", invocation, future.cause());
                                    }
                                }
                            });
                }
            });
            RpcContext.setFuture(rpcFuture);
//            rpcFuture.thenApplyAsync(new Function<Object, Object>() {
//                @Override
//                public Object apply(Object o) {
//                    System.out.println("123124");
//                    System.out.println(o);
//                    return o;
//                }
//            });
            return rpcFuture;
        } catch (Exception e) {
            log.error("Client Request Error:", e.getMessage(), e);
            throw new RuntimeException("Client Error");
        }
    }
}
